
#include "db/compact_impl.hxx"
#include "db/dbenv_impl.hxx"
#include "util/common.hxx"
#include "util/logger.hxx"


Compact* CompactImpl::m_instance = NULL;

void * CompactImpl::compactThreadFunc(void* arg) {

    CompactImpl* compactImp = (CompactImpl*)arg;
    
    DbStatus ret;

    //for interval
    bool shouldSleep = false;
    
    while(true) {
    
        pthread_mutex_lock(&compactImp->m_mutex);

        while(!compactImp->m_option.m_quit && 
                (!compactImp->m_option.m_enabled || shouldSleep) ) {

            utility::thread_cond_timewait(&compactImp->m_cond, &compactImp->m_mutex, compactImp->m_interval);
            shouldSleep = false;
        }

        pthread_mutex_unlock(&compactImp->m_mutex);

        if(compactImp->m_option.m_quit) break;

        Logger::log("compact thread is working");
        ret = compactImp->m_env->compact(compactImp->m_option);
        if(!ret.isok()) Logger::log("compact thread gets error: %s",
                ret.msg().c_str());

        Logger::log("compact thread finish one round");

        shouldSleep = true;
    }

    pthread_mutex_lock(&compactImp->m_mutex);
    compactImp->m_alreadyQuit = true;
    pthread_cond_signal(&compactImp->m_cond);
    pthread_mutex_unlock(&compactImp->m_mutex);
    
    return NULL;
}


CompactImpl::CompactImpl(DbEnvImpl& env) 
: m_env(&env), m_alreadyQuit(false) {

}

CompactImpl::~CompactImpl() {

    pthread_mutex_destroy(&m_mutex);
    pthread_cond_destroy(&m_cond);
}

DbStatus CompactImpl::init() {

    DbStatus ret;

    //set default vlaue
    m_option.m_enabled = false;
    m_option.m_validDataRatio = 0.5;
    m_option.m_quit = false;
    m_option.m_segs = 0;

    m_interval = 10;

    //initialize mutex and cond
    int rc = pthread_mutex_init(&m_mutex, NULL);
    if(rc != 0) return DbStatus("CompactImpl::init::pthread_mutex_init", rc);

    rc = pthread_cond_init(&m_cond, NULL);
    if(rc != 0) return DbStatus("CompactImpl::init::pthread_cond_init", rc);

    //start thread
    utility::startThread(
            reinterpret_cast<utility::THREADFUNC*>(&CompactImpl::compactThreadFunc), (void*)this);

    return ret;
}

void CompactImpl::start() {

    pthread_mutex_lock(&m_mutex);
    m_option.m_enabled = true;
    pthread_cond_signal(&m_cond);
    pthread_mutex_unlock(&m_mutex);
}


void CompactImpl::stop(bool quit) {

    pthread_mutex_lock(&m_mutex);
    
    m_option.m_enabled = false;
    if(quit) m_option.m_quit = true;

    pthread_cond_signal(&m_cond);

    while(quit && !m_alreadyQuit) {
        pthread_cond_wait(&m_cond, &m_mutex);
    }

    pthread_mutex_unlock(&m_mutex);
}


void CompactImpl::enableThrottling() {

   m_option.m_throttling.enable(); 
}


void CompactImpl::disableThrottling() {

    m_option.m_throttling.disable();
}

void CompactImpl::setThrottling(uint32_t M, uint32_t N) {

    m_option.m_throttling.setMN(M, N);
}


bool  CompactImpl::isEnabled() {
 
    return m_option.m_enabled;
}

void CompactImpl::setValidDataRatio(double ratio) {

    m_option.m_validDataRatio = ratio;
}

double  CompactImpl::getValidDataRatio() {

    return  m_option.m_validDataRatio;
}


void CompactImpl::notify() {

    if( m_option.m_enabled) pthread_cond_signal(&m_cond);
}


uint64_t CompactImpl::timeToWait() {

    if(m_option.shouldDone() || !m_option.m_compacting) return 0;

    uint64_t wt = 0;

    if(m_option.m_segs > 10) wt = 1000000;
    else if(m_option.m_segs > 2) {
    
        wt = m_option.m_segs * 100000; 
    }

    return wt;
}

bool CompactIndexCompartor::operator()(const CompactIndexItem& left, const CompactIndexItem& right) {
    return left.m_addr < right.m_addr;
}

CompactIndexItem::CompactIndexItem(uint32_t index, uint64_t addr) 
: m_index(index), 
  m_addr(addr) {
}


Compact* CompactImpl::getInstance() {

    return m_instance;
}

Compact* CompactImpl::getInstance(DbEnvImpl& env) {

    m_instance = new CompactImpl(env);

    return m_instance;
}
